package com.learn.kafka.serializer;

import com.learn.kafka.config.ConsumerProperties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;

import java.time.Duration;
import java.util.Arrays;

/**
 * 处理字节数组反序列化的消费者（继承通用配置父类）
 * @author: learn.kafka
 */
public class SerializerConsumer extends ConsumerProperties {

    public static void main(String[] args) {
        new SerializerConsumer().startMessages();
    }

    // 构造方法：初始化当前消费者的特殊配置
    public SerializerConsumer() {
        super(); // 调用父类构造方法，加载通用配置
        initConsumerConfig();
    }

    // 初始化当前消费者的特殊配置（覆盖父类默认值）
    private void initConsumerConfig() {
        // 设置消费组 ID（覆盖父类默认的 "default-group"）
        setProperty("group.id", "mate-ser-group");
        // 禁用自动偏移量重置（覆盖父类的 "earliest"）
//        setProperty("auto.offset.reset", ""); // 原代码中该配置被注释，此处清空
        // 关键：使用 ByteArrayDeserializer 处理 value（覆盖父类的 StringDeserializer）
        setProperty("value.deserializer", ByteArrayDeserializer.class.getName());
    }

    // 启动消费逻辑：接收字节数组并转换为 User 对象
    private void startMessages() {
        // 从父类获取消费者实例（泛型为 <String, byte []>，匹配字节数组类型）
        Consumer<Integer, byte[]> consumer = new KafkaConsumer<Integer, byte[]>(super.getProps());
        // 订阅目标主题
        consumer.subscribe(Arrays.asList("ser-topic"));
        try {
            while (true) {
                ConsumerRecords<Integer, byte[]> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<Integer, byte[]> record : records) {
                    // 将字节数组转换为 User 对象并输出
                    User user = new User(record.value()); // 假设 User 类有接收 byte [] 的构造方法

                    // 正确写法（移除 %s 前的空格）
                    System.out.printf("offset = %d, key = %s, value = %s, partition = %d%n",
                            record.offset(), record.key(), user, record.partition());
                }
            }
        } finally {
            // 关闭消费者，释放资源
            consumer.close();
        }

    }
}
